package bch.kafka.integrate.stream;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.*;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.time.Duration;

@Component
public class UserRiskStream {
    public static final String EVENT = "event";
    public static final String USER_ID = "userId";
    public static final String USER_EVENTS_TOPIC = "user-events";
    public static final String USER_EVENTS_LOGIN_FAIL = "login_fail";

    private static final ObjectMapper objectMapper = new ObjectMapper();

    @Bean
    public KStream<String, String> kStream(StreamsBuilder builder) {
        KStream<String, String> stream = builder.stream(USER_EVENTS_TOPIC, Consumed.with(Serdes.String(), Serdes.String()));

        stream.filter((key, value) -> {
                    try {
                        JsonNode node = objectMapper.readTree(value);
                        return USER_EVENTS_LOGIN_FAIL.equals(node.get(EVENT).asText());
                    } catch (Exception e) {
                        return false;
                    }
                })
                // 按用户分组
                .selectKey((key, value) -> {
                    try {
                        JsonNode node = objectMapper.readTree(value);
                        return node.get(USER_ID).asText();
                    } catch (Exception e) {
                        return key;
                    }
                })
                .groupByKey(Grouped.with(Serdes.String(), Serdes.String()))
                // 5分钟滑动窗口
                .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
                // 聚合计数
                .count()
                // 过滤 >=3 次的告警
                .toStream()
                .filter((windowedUserId, count) -> count >= 3)
                .peek((windowedUserId, count) ->
                        System.out.println("告警: 用户 " + windowedUserId.key() + " 5分钟内登录失败 " + count + " 次"))
        ;
//                .to("risk-alerts", Produced.with(Serdes.String(), Serdes.String()));

        return stream;
    }
}
